package com.Reactor3参考文档;

import java.time.Duration;
import java.util.Random;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.LongAdder;

import org.junit.Test;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

// 在你了解错误处理操作符之前，你必须牢记 响应式流中的任何错误都是一个终止事件。 即使用了错误处理操作符，也不会让源头流序列继续。而是将 onError 信号转化为一个 新的 序列 的开始。换句话说，它代替了被终结的 上游 流序列。
public class Error {

	private static final Logger log = LoggerFactory.getLogger(Error.class);

	private String doSomethingDangerous(Integer i) {
		return "hello: " + i;
	}

	@Test
	// 与第 (1) 条（捕获并返回一个静态的缺省值）对应的是 onErrorReturn
	public void test() {
		Flux.just(10).map(this::doSomethingDangerous).onErrorReturn("RECOVERED");
	}

	@Test
	// 你还可以通过判断错误信息的内容，来筛选哪些要给出缺省值，哪些仍然让错误继续传递下去
	public void test2(){
		Flux.just(10).map(this::doSomethingDangerous)
		.onErrorReturn(e -> e.getMessage().equals("boom10"), "recovered10");
	}

	@Test
	public void test3(){
		Flux.just("key1", "key2")
		// 对于每一个 key， 异步地调用一个外部服务
	    .flatMap(k -> callExternalService(k))
	    // 如果对外部服务的调用失败，则再去缓存中查找该 key。注意，这里无论 e 是什么，都会执行异常处理方法
	    .onErrorResume(e -> getFromCache(e));
	}

	private Publisher<String> getFromCache(Throwable e) {
		return null;
	}

	private Publisher<String> callExternalService(String k) {
		return Mono.just(k);
	}

	@Test
	// 这个函数式允许开发者自行决定如何处理
	public void test4(){
		Flux.just("timeout1", "unknown", "key2")
	    .flatMap(k -> callExternalService(k))
	    .onErrorResume(error -> {
	    	// 如果源超时，使用本地缓存
	        if (error instanceof TimeoutException)
	            return getFromCache(error);
	        // 如果源找不到对应的 key，创建一个新的实体
	        else if (error instanceof IllegalArgumentException)
	            return registerNewEntry(1, "DEFAULT");
	        else
	        	// 否则， 将问题“重新抛出”
	            return Flux.error(error);
	    });
	}

	private Publisher<String> registerNewEntry(int i, String string) {
		return null;
	}

	@Test
	// 在“错误处理方法”的例子中，基于 flatMap 方法的最后一行，我们可以猜到如何做到第 (4) 条（捕获，包装到一个业务相关的异常，然后抛出业务异常）：
	public void test5(){
		Flux.just("timeout1")
	    .flatMap(k -> callExternalService(k))
	    .onErrorResume(original -> Flux.error(
	        new BusinessException("oops, SLA exceeded", original)
	    ));
	}

	@Test
	public void test6(){
		Flux.just("timeout1")
	    .flatMap(k -> callExternalService(k))
	    .onErrorMap(original -> new BusinessException("oops, SLA exceeded", original));
	}

	@Test
	// 如果对于错误你只是想在不改变它的情况下做出响应（如记录日志），并让错误继续传递下去， 那么可以用 doOnError 方法。这对应第 (5) 条（捕获，记录错误日志，并继续抛出）。 这个方法与其他以 doOn 开头的方法一样，只起副作用（"side-effect"）。它们对序列都是只读， 而不会带来任何改动。
	// 如下边的例子所示，我们会记录错误日志，并且还通过变量自增统计错误发生个数。
	public void test7(){
		LongAdder failureStat = new LongAdder();
		Flux<String> flux = Flux.just("unknown")
		    .flatMap(k -> callExternalService(k))
		    .doOnError(e -> {
		        failureStat.increment();
		        log("uh oh, falling back, service failed for key " + e);
		    })
		    .onErrorResume(e -> getFromCache(e));
	}

	// 记录日志
	private void log(String string) {

	}

	@Test
	// 演示终止方法 onError
	// 为了演示当错误出现的时候如何导致上游序列终止，我们使用 Flux.interval 构造一个更加直观的例子。 这个 interval 操作符会在每 x 单位的时间发出一个自增的 Long 值。
	public void test8() throws InterruptedException{
		Flux<String> flux =
		Flux.interval(Duration.ofMillis(250))
		    .map(input -> {
		        if (input < 3) return "tick " + input;
		        throw new RuntimeException("boom");
		    })
		    .onErrorReturn("Uh oh");

		flux.subscribe(System.out::println);
		// 注意 interval 默认基于一个 timer Scheduler 来执行。 如果我们想在 main 方法中运行， 我们需要调用 sleep，这样程序就可以在还没有产生任何值的时候就退出了。
		// 即使多给了 1 秒钟时间，也没有更多的 tick 信号由 interval 产生了，所以序列确实被错误信号终止了
		Thread.sleep(2100);
	}

	@Test
	public void test9() {
		Flux.just("user-1") // 生成一个用户流，这些用户会请求自己的电影推荐
			.flatMap(user -> // 针对每个用户，调用不可靠的recommendedBooks服务，如果失败，将以指数退避重试(不超过5次，从100毫秒的持续时间开始)。但是如果重试策略在3s没有带来任何结果，则会触发一个错误信号
					recommendedBooks(user)
					.retryBackoff(5, Duration.ofMillis(100))
					.timeout(Duration.ofSeconds(3))
					.onErrorResume(e -> Flux.just("the martian"))) // 最后，如果出现任何错误，使用onErrorResume操作符返回预定于的通用推荐集
			.subscribe(
					b -> log.info("onNext: {}", b),
					e -> log.warn("onError: {}", e.getMessage()),
					() -> log.info("onComplate")
			);
	}

	private static final Random random = new Random();

	private Flux<String> recommendedBooks(String userId){
		return Flux.defer(() ->{ // 延迟计算，直到订阅者达到
			if(random.nextInt(10) < 7) { // 我们的不可靠服务很可能返回错误，但是，我们可以通过应用delaySequence操作符来及时转移所有信号
				return Flux.<String>error(new RuntimeException("err")).delaySequence(Duration.ofMillis(100));
			}else { // 如果客户端很幸运，他会收到一些延迟了的推荐
				return Flux.just("blue mars", "the expanse").delayElements(Duration.ofMillis(50));
			}
		}).doOnSubscribe(s -> log.info("request for {}", userId)); // 将请求记录到服务中
	}

}
